﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Text.RegularExpressions;
using System.Data;
using System.Data.Odbc;
using System.Data.SqlClient;
using System.IO;
using System.Xml;
using System.Threading;
using MySql.Data.MySqlClient;

using MyLibrary.Text.Match;
using MyLibrary.IO;
using MyLibrary.Tools;
using FTP_COMLib;

//下限探测读取法:在读取的记录数未知的情况下的一种读取方式,设置的每次的读取量.如果某次的读取量少于规定的读取量.读取结束
//该方法用于读取,在计算某类记录的数据量需要大量时间的查询时使用

namespace MyLibrary.Data.Transmission
{
    /// <summary>
    /// 数据类型
    /// </summary>
    public enum DatabaseType
    {
        /// <summary>
        /// FTP服务器上的文件
        /// </summary>
        FTPFile,
        /// <summary>
        /// MySQL数据库
        /// </summary>
        MySQL,
        /// <summary>
        /// SQL Server 数据库
        /// </summary>
        MSSQL,
        /// <summary>
        /// 使用ODBC连接方式
        /// </summary>
        ODBC
    }

    /// <summary>
    /// 媒介类型
    /// </summary>
    public enum MediaType
    {
        /// <summary>
        /// 一般文本文件
        /// </summary>
        NormalFile,
        /// <summary>
        /// Xml文件格式
        /// </summary>
        XmlFile,
        /// <summary>
        /// 直接插入
        /// </summary>
        Direct,
        /// <summary>
        /// 数据库中转XML文件
        /// </summary>
        Database
    }

    /// <summary>
    /// 传输类型
    /// </summary>
    public enum TransportType
    {
        /// <summary>
        /// 重新装载(删除源数据表中的所有记录)
        /// </summary>
        ReLoad,
        /// <summary>
        /// 增量添加
        /// </summary>
        Increase,
        /// <summary>
        /// 每日添加
        /// </summary>
        Daily,
        /// <summary>
        /// 区间
        /// </summary>
        Range,
        /// <summary>
        /// 数据操作
        /// </summary>
        Operate
    }

    /// <summary>
    /// 操作类型
    /// </summary>
    public enum OperateType
    {
        /// <summary> 建立索引 </summary>
        CreateIndex,
        /// <summary> 删除索引 </summary>
        DropIndex,
        /// <summary> 建立数据表 </summary>
        CreateTable,
        /// <summary> 删除数据表 </summary>
        DropTable,
        /// <summary> 建立视图 </summary>
        CreateView,
        /// <summary> 删除视图 </summary>
        DropView
    }

    /// <summary>
    /// 执行操作面对的数据库
    /// </summary>
    public enum TargetDatabase
    {
        /// <summary>
        /// 源数据库
        /// </summary>
        Source,
        /// <summary>
        /// 目标数据库
        /// </summary>
        Targe
    }

    /// <summary>
    /// 设置基类
    /// </summary>
    public abstract class Setting
    {
        public abstract string Id { get; }
        protected abstract void LoadConfig(string configFile, string configSection);
        protected abstract void LoadConfig(XmlNode configRoot);
    }

    /// <summary>
    /// 操作配置
    /// </summary>
    public class OperateSetting : Setting
    {
        #region 属性
        private string _id = String.Empty;
        /// <summary>
        /// 配置节点的ID[只读]
        /// </summary>
        public override string Id
        {
            get { return _id; }
        }
        /// <summary>
        /// 操作类型
        /// </summary>
        private OperateType _operateType;
        /// <summary>
        /// 操作类型[只读]
        /// </summary>
        public OperateType OperateType
        {
            get { return _operateType; }
        }
        /// <summary>
        /// 操作对象的名称[只读]
        /// </summary>
        private string _objectName = String.Empty;
        public string ObjectName
        {
            get { return _objectName; }
        }
        private TargetDatabase _targetDB = TargetDatabase.Source;
        /// <summary>
        /// 操作面向的数据库
        /// </summary>
        public TargetDatabase TargetDatabase
        {
            get { return _targetDB; }
        }
        /// <summary>
        /// 建立索引设置[只读]
        /// </summary>
        private IndexParameters _indexParameters = null;
        public IndexParameters IndexAttributes
        {
            get { return _indexParameters; }
        }
        private TableParameters _tableParameters = null;
        /// <summary>
        /// 建立数据表设置[只读]
        /// </summary>
        public TableParameters TableAttributes
        {
            get { return _tableParameters; }
        }
        #endregion

        #region 构造函数
                /// <summary>
        /// 通过配置文件指定传输设置
        /// </summary>
        /// <param name="configFile">配置文件路径</param>
        /// <param name="configSection">配置节的XPath</param>
        public OperateSetting(string configFile, string configSection)
        {
            if (String.IsNullOrEmpty(configFile)) throw new ArgumentNullException("配置文件名不能未空");
            if (String.IsNullOrEmpty(configSection)) throw new ArgumentNullException("配置节点名不能为空");
            if (!File.Exists(configFile)) throw new FileNotFoundException("指定的配置文件不存在");
            //读取配置
            this.LoadConfig(configFile, configSection);
        }

        /// <summary>
        /// 初始化默认状态的传输配置
        /// </summary>
        public OperateSetting()
        {
        }

        /// <summary>
        /// 通过配置节点得到传输配置
        /// </summary>
        /// <param name="configNode">配置节点</param>
        public OperateSetting(XmlNode configNode)
        {
            if (configNode == null) throw new ArgumentNullException("配置节点不能为空");
            if (configNode.ChildNodes.Count == 0) throw new ArgumentNullException("不是有效的配置节点");

            //读取配置
            this.LoadConfig(configNode);
        }
        #endregion

        #region 重写的方法
        protected override void LoadConfig(string configFile, string configSection)
        {
            XmlNode configRoot = CommonTools.GetConfigNode(configSection, configFile);
            LoadConfig(configRoot);
        }
        protected override void LoadConfig(XmlNode configRoot)
        {
            if (configRoot == null) throw new XmlException("无法读取指定的配置节");
            if (configRoot.Attributes["id"] == null) throw new ArgumentException("配置节点的ID属性不存在");
            if (configRoot.Attributes["operate"] == null) throw new ArgumentException("配置节点的operate属性不存在");
            //ID
            this._id = configRoot.Attributes["id"].Value;
            //获得操作类型
            string tmp = String.Empty;
            tmp = configRoot.Attributes["operate"].Value.Trim().ToLower();
            if (tmp == "cindex")
                this._operateType = OperateType.CreateIndex;
            else if (tmp == "dindex")
                this._operateType = OperateType.DropIndex;
            else if (tmp == "ctable")
                this._operateType = OperateType.CreateTable;
            else if (tmp == "dtable")
                this._operateType = OperateType.DropTable;
            //else if (tmp == "cview")
            //    this._operateType = OperateType.CreateView;
            //else if (tmp == "dview")
            //    this._operateType = OperateType.DropView;
            else
                throw new NotSupportedException("设置了不支持的操作");

            if (configRoot["ObjectName"] == null) throw new ArgumentNullException("没有指定操作的对象名称");
            this._objectName = configRoot["ObjectName"].InnerText;

            if (configRoot["ObjectName"].Attributes["target"] == null) throw new ArgumentNullException("没有执行操作面向的数据库");
            if (configRoot["ObjectName"].Attributes["target"].Value.ToLower() == "target")
                this._targetDB = TargetDatabase.Targe;
            else
                this._targetDB = TargetDatabase.Source;

            //操作类型为建立索引时
            if (this._operateType == OperateType.CreateIndex)
            {
                if (configRoot["ObjectName"].ChildNodes.Count == 0) throw new ArgumentException("建立索引时至少需要指定一个字段");
                if (configRoot["ObjectName"].Attributes["table"] == null) throw new ArgumentException("必须指定建立索引的数据表");
                if (configRoot["ObjectName"].Attributes["name"] == null) throw new ArgumentException("必须执行要建立索引的名称");
                _indexParameters = IndexParameters.GetInstance();

                this._objectName = configRoot["ObjectName"].Attributes["name"].Value;
                if (configRoot["ObjectName"].Attributes["unique"] != null)
                {
                    if (configRoot["ObjectName"].Attributes["unique"].Value.ToLower() == "true")
                        this._indexParameters.IsClustered = true;
                    else
                        this._indexParameters.IsClustered = false;
                }
                _indexParameters.TableName = configRoot["ObjectName"].Attributes["table"].Value;
                if (configRoot["ObjectName"].Attributes["store"] != null)
                    this._indexParameters.StorePattern = configRoot["ObjectName"].Attributes["store"].Value;
                else
                    this._indexParameters.StorePattern = "Primary";
                foreach (XmlNode node in configRoot["ObjectName"].ChildNodes)
                    this._indexParameters.IndexColumns.Add(node.InnerText);
            }
            else if (this._operateType == OperateType.CreateTable)
            {
                if (configRoot["ObjectName"].ChildNodes.Count == 0) throw new ArgumentException("建立表时至少需要指定一个字段");
                _tableParameters = TableParameters.GetInstance();

                if (configRoot["ObjectName"].Attributes["store"] != null)
                    this._indexParameters.StorePattern = configRoot["ObjectName"].Attributes["store"].Value;
                else
                    this._indexParameters.StorePattern = "Primary";

                foreach (XmlNode node in configRoot["ObjectName"].ChildNodes)
                    this._tableParameters.TableColumns.Add(node.InnerText, node.Attributes["type"].Value);
            }
        }
        #endregion

        /// <summary>
        /// 建立索引需要的参数
        /// </summary>
        public class IndexParameters
        {
            private bool _isClustered = false;
            /// <summary>
            /// 是否为聚集索引
            /// </summary>
            public bool IsClustered
            {
                get { return this._isClustered; }
                set { this._isClustered = false; }
            }
            private string _storePattern = String.Empty;
            /// <summary>
            /// 存放格局
            /// </summary>
            public string StorePattern
            {
                get { return this._storePattern; }
                set { this._storePattern = value; }
            }
            private string _tableName = String.Empty;
            /// <summary>
            /// 使用这个索引的表名称
            /// </summary>
            public string TableName
            {
                get { return this._tableName; }
                set { this._tableName = value; }
            }
            private List<string> _indexColumns = new List<string>();
            public List<string> IndexColumns
            {
                get { return this._indexColumns; }
                set { this._indexColumns = value; }
            }

            private static IndexParameters _instance = null;
            private IndexParameters()
            {
            }
            public static IndexParameters GetInstance()
            {
                if (_instance != null)
                {
                    _instance._isClustered = false;
                    _instance._storePattern = String.Empty;
                    _instance._indexColumns = new List<string>();
                }
                _instance = new IndexParameters();
                return _instance;
            }
        }
        /// <summary>
        /// 建立数据表需要的参数
        /// </summary>
        public class TableParameters
        {
            private string _storePattern = String.Empty;
            /// <summary>
            /// 存放格局
            /// </summary>
            public string StorePattern
            {
                get { return this._storePattern; }
                set { this._storePattern = value; }
            }
            private Dictionary<string, string> _tableColumns = new Dictionary<string, string>();
            public Dictionary<string, string> TableColumns
            {
                get { return this._tableColumns; }
                set { this._tableColumns = value; }
            }

            private static TableParameters _instance = null;
            private TableParameters()
            {
            }
            public static TableParameters GetInstance()
            {
                if (_instance != null)
                {
                    _instance._storePattern = String.Empty;
                    _instance._tableColumns = new Dictionary<string, string>();
                }
                _instance = new TableParameters();
                return _instance;
            }
        }
    }

    /// <summary>
    /// 传输配置
    /// </summary>
    public class TransportSetting : Setting
    {
        #region 属性
        private string _id = String.Empty;
        /// <summary>
        /// 传输设置的唯一标示
        /// </summary>
        public override string Id
        {
            get { return _id; }
        }
        private TransportType _transportType = TransportType.ReLoad;
        /// <summary>
        /// 传输数据的类型
        /// </summary>
        public TransportType TransportType
        {
            get { return _transportType; }
            set { _transportType = value; }
        }
        private MediaType _mediaType = MediaType.Direct;
        /// <summary>
        /// 传输数据的媒介
        /// </summary>
        public MediaType MediaType
        {
            get { return _mediaType; }
            set { _mediaType = value; }
        }
        private string _sourceTable = String.Empty;
        /// <summary>
        /// 源数据表
        /// </summary>
        public string SourceTable
        {
            get { return _sourceTable; }
            set { _sourceTable = value; }
        }
        private string _targetTable = String.Empty;
        /// <summary>
        /// 目标数据表
        /// </summary>
        public string TargetTable
        {
            get { return _targetTable; }
            set { _targetTable = value; }
        }
        private int _threadNumber = -1;
        /// <summary>
        /// 使用的线程数量
        /// </summary>
        public int ThreadNumber
        {
            get { return _threadNumber; }
            set { _threadNumber = value; }
        }
        private string _filePath = String.Empty;
        /// <summary>
        /// 文件存放路径
        /// </summary>
        public string FilePath
        {
            get { return _filePath; }
            set { _filePath = value; }
        }
        private string _prepareCommand = String.Empty;
        /// <summary>
        /// 插入数据表前的预处理语句
        /// </summary>
        public string PrepareCommand
        {
            get { return _prepareCommand; }
            set { _prepareCommand = value; }
        }
        private string _loadCommand = String.Empty;
        /// <summary>
        /// 读取记录的语句
        /// </summary>
        public string LoadCommand
        {
            get { return _loadCommand; }
            set { _loadCommand = value; }
        }

        private Dictionary<string, string> _loadCommands = null;
        /// <summary>
        /// 多重执行语句
        /// </summary>
        public Dictionary<string, string> LoadCommands
        {
            get { return _loadCommands; }
            set { _loadCommands = value; }
        }

        private string _insertCommand = String.Empty;
        /// <summary>
        /// 插入数据的语句
        /// </summary>
        public string InsertCommand
        {
            get { return _insertCommand; }
            set { _insertCommand = value; }
        }

        private string _exportCommand = String.Empty;
        /// <summary>
        /// 传导语句
        /// </summary>
        public string ExportCommand
        {
            get { return _exportCommand; }
            set { _exportCommand = value; }
        }
        private string _getCountCommand = String.Empty;
        /// <summary>
        /// 获取记录数量的语句
        /// </summary>
        public string GetCountCommand
        {
            get { return _getCountCommand; }
            set { _getCountCommand = value; }
        }
        private int _numberOfReadOnce = -1;
        /// <summary>
        /// 线程一次处理的记录数
        /// </summary>
        public int NumberOfReadOnce
        {
            get { return _numberOfReadOnce; }
            set { _numberOfReadOnce = value; }
        }

        private string _dateFormat;
        /// <summary>
        /// 日期格式
        /// </summary>
        public string DateFormat
        {
            get { return _dateFormat; }
            set { _dateFormat = value; }
        }

        private string _beginDate;
        /// <summary>
        /// 开始时间
        /// </summary>
        public string BeginDate
        {
            get { return _beginDate; }
            set { _beginDate = value; }
        }

        private string _endDate;
        /// <summary>
        /// 结束时间
        /// </summary>
        public string EndDate
        {
            get { return _endDate; }
            set { _endDate = value; }
        }

        private int _delayDate = 0;
        //延迟天数
        public int DelayDate
        {
            get { return _delayDate; }
            set { _delayDate = value; }
        }
        private DateTime _validateTimeStamp = DateTime.MinValue;
        /// <summary>
        /// 控制时间戳
        /// </summary>
        public DateTime ValidateTimeStamp
        {
            get { return _validateTimeStamp; }
            set 
            {
                if(value != DateTime.MinValue)
                {
                    if(this._prepareCommand != String.Empty)
                        this._prepareCommand = this._prepareCommand.Replace("#timestamp#", value.ToString("yyyy-MM-dd"));
                    if(this._loadCommand != String.Empty)
                        this._loadCommand = this._loadCommand.Replace("#timestamp#", value.ToString("yyyy-MM-dd"));
                    if(this._insertCommand != String.Empty)
                        this._insertCommand = this._insertCommand.Replace("#timestamp#", value.ToString("yyyy-MM-dd"));
                    if (this._loadCommands != null)
                    {
                        Dictionary<string, string>.KeyCollection.Enumerator enumetator = this._loadCommands.Keys.GetEnumerator();
                        List<string> keys = new List<string>();
                        while (enumetator.MoveNext())
                            keys.Add(enumetator.Current);
                        foreach(string key in keys)
                            this._loadCommands[key] = this._loadCommands[key].Replace("#timestamp#", value.ToString("yyyy-MM-dd"));
                    }
                    this._validateTimeStamp = value;
                }
            }
        }
        private int _minValue;
        /// <summary>
        /// 最小值 
        /// </summary>
        public int MinValue
        {
            get { return _minValue; }
            set { _minValue = value; }
        }
        private int _maxValue;
        /// <summary>
        /// 最大值 
        /// </summary>
        public int MaxValue
        {
            get { return _maxValue; }
            set { _maxValue = value; }
        }
        private int _subStepValue;
        /// <summary>
        /// 二级步长值 
        /// </summary>
        public int SubStepValue
        {
            get { return _subStepValue; }
            set { _subStepValue = value; }
        }
        private int _stepMaxValue;
        /// <summary>
        /// 步长的最大值 
        /// </summary>
        public int SubMaxValue
        {
            get { return _stepMaxValue; }
            set { _stepMaxValue = value; }
        }
        private int _stepValue;
        /// <summary>
        /// 步长
        /// </summary>
        public int StepValue
        {
            get { return _stepValue; }
            set { _stepValue = value; }
        }
        #endregion 

        /// <summary>
        /// 通过配置文件指定传输设置
        /// </summary>
        /// <param name="configFile">配置文件路径</param>
        /// <param name="configSection">配置节的XPath</param>
        public TransportSetting(string configFile, string configSection)
        {
            if (String.IsNullOrEmpty(configFile)) throw new ArgumentNullException("配置文件名不能未空");
            if (String.IsNullOrEmpty(configSection)) throw new ArgumentNullException("配置节点名不能为空");
            if (!File.Exists(configFile)) throw new FileNotFoundException("指定的配置文件不存在");
            //读取配置
            this.LoadConfig(configFile, configSection);
        }

        /// <summary>
        /// 初始化默认状态的传输配置
        /// </summary>
        public TransportSetting()
        {
        }

        /// <summary>
        /// 通过配置节点得到传输配置
        /// </summary>
        /// <param name="configNode">配置节点</param>
        public TransportSetting(XmlNode configNode)
        {
            if (configNode == null) throw new ArgumentNullException("配置节点不能为空");
            if (configNode.ChildNodes.Count == 0) throw new ArgumentNullException("不是有效的配置节点");

            //读取配置
            this.LoadConfig(configNode);
        }

        /// <summary>
        /// 读取配置
        /// </summary>
        /// <param name="configFile">配置文件路径</param>
        /// <param name="configSection">配置节点</param>
        protected override void LoadConfig(string configFile, string configSection)
        {
            XmlNode configRoot = CommonTools.GetConfigNode(configSection, configFile);
            LoadConfig(configRoot);
        }

        /// <summary>
        /// 读取配置
        /// </summary>
        protected override void LoadConfig(XmlNode configRoot)
        {
            if (configRoot == null) throw new XmlException("无法读取指定的配置节");
            if (configRoot.Attributes["id"] == null) throw new ArgumentException("配置节点的ID属性不存在");
            if (configRoot.Attributes["type"] == null) throw new ArgumentException("配置节点的type属性不存在");
            if (configRoot.Attributes["media"] == null) throw new ArgumentException("配置节点的media属性不存在");

            //ID
            this._id = configRoot.Attributes["id"].Value;
            //类型
            string tmp = String.Empty;
            tmp = configRoot.Attributes["type"].Value.Trim().ToLower();
            if (tmp == "reload")
                this._transportType = TransportType.ReLoad;
            else if (tmp == "increase")
                this._transportType = TransportType.Increase;
            //如果是每日数据同步
            else if (tmp == "daily")
            {
                this._transportType = TransportType.Daily;
                this._dateFormat = configRoot["DateFormat"].InnerText;
                if (configRoot["DateFormat"].Attributes["delay"] != null)
                    Int32.TryParse(configRoot["DateFormat"].Attributes["delay"].Value, out this._delayDate);

                if (configRoot["BeginDate"] != null)
                    this._beginDate = configRoot["BeginDate"].InnerText;
                if (configRoot["EndDate"] != null)
                    this._endDate = configRoot["EndDate"].InnerText;
            }
            else if (tmp == "range")
            {
                this._transportType = TransportType.Range;
                if(configRoot["MinValue"] != null) this._minValue = Convert.ToInt32(configRoot["MinValue"].InnerText);
                if (configRoot["MaxValue"] != null) this._maxValue = Convert.ToInt32(configRoot["MaxValue"].InnerText);
                if (configRoot["Step"] != null)
                {
                    this._stepValue = Convert.ToInt32(configRoot["Step"].InnerText);
                    if (configRoot["Step"].Attributes["substep"] != null && configRoot["Step"].Attributes["max"] != null)
                    {
                        this._subStepValue = Convert.ToInt32(configRoot["Step"].Attributes["substep"].Value);
                        this._stepMaxValue = Convert.ToInt32(configRoot["Step"].Attributes["max"].Value);
                    }
                }
            }
            else
                this._transportType = TransportType.ReLoad;
            //传输数据的方式
            tmp = configRoot.Attributes["media"].Value.Trim().ToLower();
            if (tmp == "file")
                this._mediaType = MediaType.NormalFile;
            else if (tmp == "xml")
                this._mediaType = MediaType.XmlFile;
            else if (tmp == "direct")
                this._mediaType = MediaType.Direct;
            else
                this._mediaType = MediaType.XmlFile;
            //源数据表
            this._sourceTable = configRoot["SourceTable"].InnerText;
            //目标数据表
            this._targetTable = configRoot["TargetTable"].InnerText;
            //如果传输方式为File和XML时
            if (this._mediaType == MediaType.NormalFile || this._mediaType == MediaType.XmlFile)
            {
                if (configRoot["FilePath"] == null) throw new XmlException("当指定传输方式为文件方式时,必须指定文件存放路径");
                this._filePath = configRoot["FilePath"].InnerText;
            }
            //读取线程数量,每次执行数量
            if (configRoot["Threads"] != null)
            {
                if (!Int32.TryParse(configRoot["Threads"].InnerText, out this._threadNumber))
                    this._threadNumber = 5;
            }
            //读取语句
            ParameterFitter fitter = new ParameterFitter();
            //读取一次读取源的数量量
            if (configRoot["ReadOneTime"] != null)
            {
                if (!Int32.TryParse(configRoot["ReadOneTime"].InnerText, out this._numberOfReadOnce))
                    this._numberOfReadOnce = 2000;
                //读取一次读取的数据量
                if (configRoot["RecordCountCommand"] != null)
                {
                    tmp = configRoot["RecordCountCommand"].InnerText;
                    if (configRoot["RecordCountCommand"].Attributes["encrypt"].Value.Trim().ToLower() == "false")
                        this._getCountCommand = tmp;
                    else
                        this._getCountCommand = CommonTools.Decrypt(tmp, this._id);
                    //格式控制
                    this._getCountCommand = fitter.Fit(this._getCountCommand);
                }
            }
            if (configRoot["LoadCommand"] != null)
            {
                tmp = configRoot["LoadCommand"].InnerText;
                if (configRoot["LoadCommand"].Attributes["encrypt"].Value.Trim().ToLower() == "false")
                    this._loadCommand = tmp;
                else
                    this._loadCommand = CommonTools.Decrypt(tmp, this._id);
                //格式控制
                this._loadCommand = fitter.Fit(this._loadCommand);
            }
            //如果是多重的查询语句
            else if (configRoot["LoadCommands"] != null)
            {
                LoadCommands = new Dictionary<string, string>();
                foreach (XmlNode lNode in configRoot["LoadCommands"].ChildNodes)
                {
                    if (lNode.Attributes["encrypt"].Value.Trim().ToLower() == "false")
                        this._loadCommands.Add(lNode.Attributes["name"].Value, fitter.Fit(lNode.InnerText));
                    else
                        this._loadCommands.Add(lNode.Attributes["name"].Value,fitter.Fit(CommonTools.Decrypt(lNode.InnerText,this._id)));
                }
            }

            //插入语句
            tmp = configRoot["InsertCommand"].InnerText;
            if (configRoot["InsertCommand"].Attributes["encrypt"].Value.Trim().ToLower() == "false")
                this._insertCommand = tmp;
            else
                this._insertCommand = CommonTools.Decrypt(tmp, this._id);
            //格式控制
            this._insertCommand = fitter.Fit(this._insertCommand);

            //传导语句
            if (configRoot["ExportCommand"] != null)
            {
                tmp = configRoot["ExportCommand"].InnerText;
                if (configRoot["ExportCommand"].Attributes["encrypt"].Value.Trim().ToLower() == "false")
                    this._exportCommand = tmp;
                else
                    this._exportCommand = CommonTools.Decrypt(tmp, this._id);
                //格式控制
                this._exportCommand = fitter.Fit(this._exportCommand);
            }

            //预处理语句
            if (configRoot["PrepareCommand"] != null)
            {
                tmp = configRoot["PrepareCommand"].InnerText;
                if (configRoot["PrepareCommand"].Attributes["encrypt"].Value.Trim().ToLower() == "false")
                    this._prepareCommand = tmp;
                else
                    this._prepareCommand = CommonTools.Decrypt(tmp, this._id);
                //格式控制
                this._prepareCommand = fitter.Fit(this._prepareCommand);
            }
        }
    }

    /// <summary>
    /// 传输基类
    /// </summary>
    public abstract class TransmissionBase
    {
        public abstract DatabaseType SrcDbType { get; set; }
        public abstract string SourceConnectionString { get; set; }
        public abstract DatabaseType TagDbType { get; set; }
        public abstract string TargetConnectionString { get; set; }
        public abstract void Begin();
    }

    /// <summary>
    /// 数据传输综合类
    /// </summary>
    public class DataTransmission : TransmissionBase, IDisposable
    {
        #region 私有变量
        private string _sourceConnectionString = String.Empty;
        private string _targetConnectionString = String.Empty;
        private string _frequencyTable = String.Empty;
        private DatabaseType _srcDbType = DatabaseType.MySQL;
        private DatabaseType _tagDbType = DatabaseType.MSSQL;
        private TransportSetting _setting = null;
        private Logger logger;
        private static string _LogPath = AppDomain.CurrentDomain.BaseDirectory + @"\Log\DataTransLog_" + DateTime.Today.ToString("yyyyMMdd") + ".log";
        #endregion

        public delegate string DataRowWritingEventHandler(DataRow row);
        public delegate void FileImportEventHandler(string file);
        public event DataRowWritingEventHandler OnDataRowWriting;
        public event FileImportEventHandler OnFileImport;

        #region 属性
        /// <summary>
        /// 执行频度控制表
        /// </summary>
        public string FrequencyTable
        {
            get { return _frequencyTable; }
            set { _frequencyTable = value; }
        }
        /// <summary>
        /// 源数据库连接字符串
        /// </summary>
        public override string SourceConnectionString
        {
            get { return _sourceConnectionString; }
            set { _sourceConnectionString = value; }
        }
        /// <summary>
        /// 源数据库类型
        /// </summary>
        public override DatabaseType SrcDbType
        {
            get { return _srcDbType; }
            set { _srcDbType = value; }
        }
        /// <summary>
        /// 目标数据库连接字符串
        /// </summary>
        public override string TargetConnectionString
        {
            get { return _targetConnectionString; }
            set { _targetConnectionString = value; }
        }
        /// <summary>
        /// 目标数据库类型
        /// </summary>
        public override DatabaseType TagDbType
        {
            get { return _tagDbType; }
            set { _tagDbType = value; }
        }

        /// <summary>
        /// 数据传输设置
        /// </summary>
        public TransportSetting Setting
        {
            get { return _setting; }
            set { _setting = value; }
        }
        #endregion

        #region 构造函数
        /// <summary>
        /// 数据传输综合类
        /// </summary>
        /// <param name="settings">传输设置</param>
        public DataTransmission(TransportSetting settings)
        {
            if (settings == null) throw new ArgumentNullException("传输配置不能为空");
            _setting = settings;
            logger = Logger.CreateLogerInstance();
            logger.LogPath = _LogPath;
        }
        /// <summary>
        /// 数据传输综合类
        /// </summary>
        public DataTransmission()
        {
            logger = Logger.CreateLogerInstance();
            logger.LogPath = _LogPath;
        }

        ~DataTransmission()
        {

        }
        #endregion

        /// <summary>
        /// 开始传输数据[类的入口点]
        /// </summary>
        public override void Begin()
        {
            //判断是否达到执行周期
            if (!CheckFrequency())
                return;

            //如果存在预处理语句
            if (!String.IsNullOrEmpty(this._setting.PrepareCommand) && this._setting.TransportType != TransportType.Daily)
            {
                SqlConnection connection = new SqlConnection();
                SqlCommand command = new SqlCommand();

            RETRYPREPARE:
                try
                {
                    connection.ConnectionString = this._targetConnectionString;
                    connection.Open();

                    command.Connection = connection;
                    command.CommandTimeout = connection.ConnectionTimeout;

                    command.CommandText = this._setting.PrepareCommand;
                    Console.WriteLine(command.CommandText);
                    command.ExecuteNonQuery();
                }
                catch (Exception ex)
                {
                    //如果准备语句错误了
                    Console.WriteLine(ex.Message);
                    if(connection.State != ConnectionState.Closed)
                        connection.Close();
                    goto RETRYPREPARE;
                }
                finally
                {
                    if (connection.State != ConnectionState.Closed)
                        connection.Close();
                    connection.Dispose();
                    command.Dispose();
                }
            }

            //使用多线程
            if (this._setting.ThreadNumber != -1)
            {
                if (this._setting.MediaType == MediaType.NormalFile || this._setting.MediaType == MediaType.XmlFile || this._setting.MediaType == MediaType.Database)
                {
                    //如果目录已经存在了
                    if(Directory.Exists(this._setting.FilePath))
                    {
                        Directory.Delete(this._setting.FilePath, true);
                        Directory.CreateDirectory(this._setting.FilePath);
                    }

                    //多线程数据区间获取方式
                    if (this._setting.TransportType == TransportType.Range)
                        ExportRangeDataToFile(this._setting.ThreadNumber);
                    else
                        ExportDataToFile(this._setting.ThreadNumber);
                    if (this._tagDbType == DatabaseType.MSSQL)
                        TransportFileToTargeDatabase();
                }
                //使用直接插入数据库的方式
                else if (this._setting.MediaType == MediaType.Direct)
                {
                    throw new NotImplementedException("多线程的直接数据导入,还未实现");
                }
            }
            //使用单线程
            else
            {
                //使用文件导入的方式
                if (this._setting.MediaType == MediaType.NormalFile || this._setting.MediaType == MediaType.XmlFile || this._setting.MediaType == MediaType.Database)
                {
                    //如果目录已经存在了
                    if (Directory.Exists(this._setting.FilePath))
                    {
                        Directory.Delete(this._setting.FilePath, true);
                        Directory.CreateDirectory(this._setting.FilePath);
                    }
                    //非下探式读取
                    if (this._setting.NumberOfReadOnce == -1)
                        ExportDataToFile();
                    //下探式读取
                    else
                    {
                        if (this._setting.TransportType == TransportType.Increase || this._setting.TransportType == TransportType.ReLoad)
                            ExportDateToFileByLowLimit();
                        else if (this._setting.TransportType == TransportType.Daily)
                            ExportDateToFileDailyByLowLimit();
                        else if (this._setting.TransportType == TransportType.Range)
                            ExportRangeDataToFileByLowLimit();
                    }
                    if (this._tagDbType == DatabaseType.MSSQL)
                    {
                        TransportFileToTargeDatabase();
                        if (this._setting.MediaType == MediaType.Database)
                            TransportTempData();
                    }
                }
                //使用直接插入数据库的方式
                else if (this._setting.MediaType == MediaType.Direct)
                {
                    if (this._setting.TransportType == TransportType.Increase || this._setting.TransportType == TransportType.ReLoad)
                        TransportDateByDirect();
                    else if (this._setting.TransportType == TransportType.Daily)
                        TransportDateDailyByDirect();
                }
            }
            //执行完成后,更新执行周期
            UpdateFrequency();
        }

        /// <summary>
        /// 使用直接数据插入方式,传输数据
        /// </summary>
        protected virtual void TransportDateByDirect()
        {
            IDbConnection srcConnection = null;
            IDbConnection tagConnection = null;
            IDataAdapter adapter = null;
            IDbCommand tagCommand = null;
            IDbCommand srcCommand = null;

            //源数据库类型连接
            if (this._srcDbType == DatabaseType.ODBC)
                srcConnection = new OdbcConnection();
            else if (this._srcDbType == DatabaseType.MySQL)
                srcConnection = new MySqlConnection();
            else if (this._srcDbType == DatabaseType.MSSQL)
                srcConnection = new SqlConnection();


            //目标数据库类型连接
            if (this._tagDbType == DatabaseType.ODBC)
            {
                tagConnection = new OdbcConnection();
                tagCommand = new OdbcCommand();
            }
            else if (this._tagDbType == DatabaseType.MySQL)
            {
                tagConnection = new MySqlConnection();
                tagCommand = new MySqlCommand();
            }
            else if (this._tagDbType == DatabaseType.MSSQL)
            {
                tagConnection = new SqlConnection();
                tagCommand = new SqlCommand();
            }

            DataSet ds = new DataSet();
            string insertCommand = this._setting.InsertCommand;

            try
            {
                srcConnection.ConnectionString = this._sourceConnectionString;
                srcConnection.Open();

                //初始化命令
                tagConnection.ConnectionString = this._targetConnectionString;
                tagConnection.Open();
                tagCommand.Connection = tagConnection;

                if (_setting.LoadCommand != String.Empty)
                {
                    //读取源表的数据
                    if (this.SrcDbType == DatabaseType.ODBC)
                    {
                        srcCommand = new OdbcCommand(_setting.LoadCommand, (OdbcConnection)srcConnection);
                        adapter = new OdbcDataAdapter((OdbcCommand)srcCommand);
                    }
                    else if (SrcDbType == DatabaseType.MySQL)
                    {
                        srcCommand = new MySqlCommand(_setting.LoadCommand, (MySqlConnection)srcConnection);
                        adapter = new MySqlDataAdapter((MySqlCommand)srcCommand);
                    }
                    else if (this.SrcDbType == DatabaseType.MSSQL)
                    {
                        srcCommand = new SqlCommand(_setting.LoadCommand, (SqlConnection)srcConnection);
                        adapter = new SqlDataAdapter((SqlCommand)srcCommand);
                    }
                    adapter.Fill(ds);

                    //将Datatable中的数据以Insert的方式添加到数据库中
                    foreach (DataRow row in ds.Tables[0].Rows)
                    {
                        //拼接语句
                        foreach (DataColumn col in ds.Tables[0].Columns)
                            insertCommand = insertCommand.Replace("#" + col.ColumnName + "#", row[col].ToString());
                        tagCommand.CommandText = insertCommand;
                        Console.WriteLine(insertCommand);
                        tagCommand.ExecuteNonQuery();
                        insertCommand = this._setting.InsertCommand;
                    }
                }
                //多条记录
                else if (_setting.LoadCommands != null)
                {
                    //读取源表的数据
                    if (this.SrcDbType == DatabaseType.ODBC)
                    {
                        srcCommand = new OdbcCommand();
                        adapter = new OdbcDataAdapter((OdbcCommand)srcCommand);
                    }
                    else if (SrcDbType == DatabaseType.MySQL)
                    {
                        srcCommand = new MySqlCommand();
                        adapter = new MySqlDataAdapter((MySqlCommand)srcCommand);
                    }
                    else if (this.SrcDbType == DatabaseType.MSSQL)
                    {
                        srcCommand = new SqlCommand();
                        adapter = new SqlDataAdapter((SqlCommand)srcCommand);
                    }
                    srcCommand.Connection = srcConnection;

                    foreach (KeyValuePair<string, string> pari in _setting.LoadCommands)
                    {
                        srcCommand.CommandText = pari.Value;
                        Console.WriteLine("读取记录的语句:ID=" + pari.Key + "Command=" + pari.Value);
                        adapter.Fill(ds);

                        //将Datatable中的数据以Insert的方式添加到数据库中
                        foreach (DataRow row in ds.Tables[0].Rows)
                        {
                            //拼接语句
                            foreach (DataColumn col in ds.Tables[0].Columns)
                                insertCommand = insertCommand.Replace("#" + col.ColumnName + "#", row[col].ToString());
                            if (tagCommand.Connection.State != ConnectionState.Open)
                                tagCommand.Connection.Open();
                            tagCommand.CommandText = insertCommand;
                            Console.WriteLine("插入语句:" + insertCommand);
                            tagCommand.ExecuteNonQuery();
                            insertCommand = this._setting.InsertCommand;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ds.Clear();
                ds.Dispose();
                tagConnection.Close();
                tagConnection.Dispose();
                srcConnection.Close();
                srcConnection.Dispose();
                tagCommand.Dispose();
            }
        }

        /// <summary>
        /// 下探式读取,使用文件方式,传输每日数据
        /// </summary>
        protected virtual void ExportDateToFileDailyByLowLimit()
        {
            DateTime lastDbDate = DateTime.Today, endDbDate = DateTime.Today;
            //如果设置了结束的时间
            if (!String.IsNullOrEmpty(_setting.EndDate))
                endDbDate = Convert.ToDateTime(this._setting.EndDate);
            if (_setting.DelayDate != 0)
                endDbDate = endDbDate.AddDays(_setting.DelayDate);

            int beginCount = 0;

            IDbConnection srcConnection = null;
            IDbConnection tagConnection = null;
            IDataAdapter adapter = null;
            IDbCommand tagCommand = null;
            IDbCommand srcCommand = null;
            DataSet ds = new DataSet();
            string getDataCommand, insertDataCommand = String.Empty;
            getDataCommand = this._setting.LoadCommand;

            //源数据库类型连接
            if (this._srcDbType == DatabaseType.ODBC)
            {
                srcConnection = new OdbcConnection();
                srcCommand = new OdbcCommand();
                adapter = new OdbcDataAdapter((OdbcCommand)srcCommand);
            }
            else if (this._srcDbType == DatabaseType.MySQL)
            {
                srcConnection = new MySqlConnection();
                srcCommand = new MySqlCommand();
                adapter = new MySqlDataAdapter((MySqlCommand)srcCommand);
            }
            else if (this._srcDbType == DatabaseType.MSSQL)
                throw new NotSupportedException("设置的方法不支持MSSQL");

            try
            {
                //优先判断是否配置开始时间
                if (!String.IsNullOrEmpty(this._setting.BeginDate))
                    lastDbDate = Convert.ToDateTime(this._setting.BeginDate);
                //获取原始目标表中最大的日期
                else
                {
                    //目标数据库类型连接
                    if (this._tagDbType == DatabaseType.ODBC)
                    {
                        tagConnection = new OdbcConnection();
                        tagCommand = new OdbcCommand();
                    }
                    else if (this._tagDbType == DatabaseType.MySQL)
                    {
                        tagConnection = new MySqlConnection();
                        tagCommand = new MySqlCommand();
                    }
                    else if (this._tagDbType == DatabaseType.MSSQL)
                    {
                        tagConnection = new SqlConnection();
                        tagCommand = new SqlCommand();
                    }
                    tagConnection.ConnectionString = this._targetConnectionString;
                    tagConnection.Open();
                    tagCommand.Connection = tagConnection;
                    tagCommand.CommandText = this._setting.PrepareCommand;
                    object objDate = tagCommand.ExecuteScalar();
                    //如果最大日期存在
                    if (!(objDate is DBNull))
                        lastDbDate = Convert.ToDateTime(objDate).AddDays(1);
                    else
                        lastDbDate = endDbDate.AddDays(-1);
                    tagConnection.Close();
                }

                srcConnection.ConnectionString = this._sourceConnectionString;
                srcCommand.Connection = srcConnection;
                string fileName = String.Empty;
                Console.WriteLine("开始时间:" + lastDbDate + "~~~结束时间:" + endDbDate);
                while (endDbDate.CompareTo(lastDbDate) > 0)
                {
                    Console.WriteLine("当前线程[" + Thread.CurrentThread.Name + "]提取数据时间:" + lastDbDate.ToString("yyyy-MM-dd"));
                    //替换掉关键字
                    if (getDataCommand.Contains("#begin#"))
                        getDataCommand = getDataCommand.Replace("#begin#", lastDbDate.ToString("yyyy-MM-dd"));
                    if (getDataCommand.Contains("#beginF#"))
                        getDataCommand = getDataCommand.Replace("#beginF#", lastDbDate.ToString(this._setting.DateFormat));
                    if (getDataCommand.Contains("#end#"))
                        getDataCommand = getDataCommand.Replace("#end#", lastDbDate.AddDays(1).ToString("yyyy-MM-dd"));
                    if (getDataCommand.Contains("#endF#"))
                        getDataCommand = getDataCommand.Replace("#endF#", lastDbDate.AddDays(1).ToString(this._setting.DateFormat));

                    while (true)
                    {
                        srcCommand.CommandText = String.Format(getDataCommand, beginCount, this._setting.NumberOfReadOnce);
                        Console.WriteLine("读取数据语句:" + srcCommand.CommandText);
                        if (srcCommand.Connection.State != ConnectionState.Open)
                            srcCommand.Connection.Open();
                        adapter.Fill(ds);

                        //将Datatable中的数据写入到文件中
                        if (_setting.MediaType == MediaType.XmlFile)
                            DataTableToFile(ds.Tables[0], _setting.FilePath);
                        else if (_setting.MediaType == MediaType.NormalFile)
                            DataTableToFile(ds.Tables[0], _setting.FilePath, ref fileName);

                        if (ds.Tables[0].Rows.Count != this._setting.NumberOfReadOnce)
                            break;
                        else
                        {
                            beginCount += this._setting.NumberOfReadOnce;
                            srcCommand.Connection.Close();
                            ds.Clear();
                        }
                    }
                    //还原配置
                    beginCount = 0;
                    getDataCommand = this._setting.LoadCommand;
                    lastDbDate = lastDbDate.AddDays(1);
                    ds.Clear();
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ds.Clear();
                ds.Dispose();
                srcConnection.Close();
                srcConnection.Dispose();
            }

        }

        /// <summary>
        /// 使用直接数据插入方式,传输每日数据
        /// </summary>
        protected virtual void TransportDateDailyByDirect()
        {
            if (String.IsNullOrEmpty(this._setting.PrepareCommand))
                return;

            DateTime lastDbDate = DateTime.Today, endDbDate = DateTime.Today;
            if (!String.IsNullOrEmpty(_setting.EndDate))
                endDbDate = Convert.ToDateTime(this._setting.EndDate);
            if (_setting.DelayDate != 0)
                endDbDate = endDbDate.AddDays(_setting.DelayDate);
                
            IDbConnection srcConnection = null;
            IDbConnection tagConnection = null;
            IDataAdapter adapter = null;
            IDbCommand tagCommand = null;
            IDbCommand srcCommand = null;

            //源数据库类型连接
            if (this._srcDbType == DatabaseType.ODBC)
            {
                srcConnection = new OdbcConnection();
                adapter = new OdbcDataAdapter();
                srcCommand = new OdbcCommand();
            }
            else if (this._srcDbType == DatabaseType.MySQL)
            {
                srcConnection = new MySqlConnection();
                adapter = new MySqlDataAdapter();
                srcCommand = new MySqlCommand();
            }
            else if (this._srcDbType == DatabaseType.MSSQL)
            {
                srcConnection = new SqlConnection();
                adapter = new SqlDataAdapter();
                srcCommand = new SqlCommand();
            }

            //目标数据库类型连接
            if (this._tagDbType == DatabaseType.ODBC)
            {
                tagConnection = new OdbcConnection();
                tagCommand = new OdbcCommand();
            }
            else if (this._tagDbType == DatabaseType.MySQL)
            {
                tagConnection = new MySqlConnection();
                tagCommand = new MySqlCommand();
            }
            else if (this._tagDbType == DatabaseType.MSSQL)
            {
                tagConnection = new SqlConnection();
                tagCommand = new SqlCommand();
            }

            DataSet ds = new DataSet();
            string getDataCommand, insertDataCommand = String.Empty;
            getDataCommand = this._setting.LoadCommand;
            insertDataCommand = this._setting.InsertCommand;

            try
            {
                tagConnection.ConnectionString = this._targetConnectionString;
                tagConnection.Open();
                tagCommand.Connection = tagConnection;
                //获取数据库目标表中最大的日期数
                tagCommand.CommandText = this._setting.PrepareCommand;
                object objDate = tagCommand.ExecuteScalar();
                //如果最大日期不为空
                if (!(objDate is DBNull))
                    lastDbDate = Convert.ToDateTime(objDate).AddDays(1);
                //如果配置了开始时间
                else if (!String.IsNullOrEmpty(this._setting.BeginDate))
                    lastDbDate = Convert.ToDateTime(this._setting.BeginDate);
                //如果最大日期为空,且没有配置开始时间,lastDbDate为结束时间的前一天
                else
                    lastDbDate = endDbDate.AddDays(-1);

                tagConnection.Close();

                srcConnection.ConnectionString = this._sourceConnectionString;
                srcCommand.Connection = srcConnection;


                while (endDbDate.CompareTo(lastDbDate) > 0)
                {
                    Console.WriteLine("当前线程[" + Thread.CurrentThread.Name + "]提取数据时间:" + lastDbDate.ToString("yyyy-MM-dd")); 
                    //替换掉关键字
                    if(getDataCommand.Contains("#begin#"))
                        getDataCommand = getDataCommand.Replace("#begin#", lastDbDate.ToString("yyyy-MM-dd"));
                    if(getDataCommand.Contains("#beginF#"))
                        getDataCommand = getDataCommand.Replace("#beginF#", lastDbDate.ToString(this._setting.DateFormat));
                    if (getDataCommand.Contains("#end#"))
                        getDataCommand = getDataCommand.Replace("#end#", lastDbDate.ToString("yyyy-MM-dd"));
                    if (getDataCommand.Contains("#endF#"))
                        getDataCommand = getDataCommand.Replace("#endF#", lastDbDate.AddDays(1).ToString(this._setting.DateFormat));

                    srcCommand.CommandText = getDataCommand;
                    //获取记录的语句
                    if (this._srcDbType == DatabaseType.ODBC)
                        ((OdbcDataAdapter)adapter).SelectCommand = (OdbcCommand)srcCommand;
                    else if (this._srcDbType == DatabaseType.MySQL)
                        ((MySqlDataAdapter)adapter).SelectCommand = (MySqlCommand)srcCommand;
                    else if (this._srcDbType == DatabaseType.MSSQL)
                        ((SqlDataAdapter)adapter).SelectCommand = (SqlCommand)srcCommand;
                    adapter.Fill(ds);

                    //将Datatable中的数据以Insert的方式添加到数据库中
                    //初始化命令
                    foreach (DataRow row in ds.Tables[0].Rows)
                    {
                        //拼接语句
                        foreach (DataColumn col in ds.Tables[0].Columns)
                            insertDataCommand = insertDataCommand.Replace("#" + col.ColumnName + "#", row[col].ToString());
                        tagCommand.CommandText = insertDataCommand;
                        if (tagCommand.Connection.State != ConnectionState.Open)
                            tagCommand.Connection.Open();

                        tagCommand.ExecuteNonQuery();
                        insertDataCommand = this._setting.InsertCommand;
                    }

                    //还原配置
                    getDataCommand = this._setting.LoadCommand;
                    lastDbDate = lastDbDate.AddDays(1);
                    ds.Clear();
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ds.Clear();
                ds.Dispose();
                tagConnection.Close();
                tagConnection.Dispose();
                srcConnection.Close();
                srcConnection.Dispose();
                tagCommand.Dispose();
            }

        }

        /// <summary>
        /// 将源数据库中的数据导出成指定的文件格式[单线程]
        /// </summary>
        protected virtual void ExportDataToFile()
        {
            IDbConnection srcConnection = null;
            IDataAdapter adapter = null;

            //源数据库类型连接
            if (this.SrcDbType == DatabaseType.ODBC)
                srcConnection = new OdbcConnection();
            else if (this.SrcDbType == DatabaseType.MySQL)
                srcConnection = new MySqlConnection();
            else if (this.SrcDbType == DatabaseType.MSSQL)
                srcConnection = new SqlConnection();
            else
                srcConnection = new OdbcConnection();

            DataSet ds = new DataSet();

            try
            {
                srcConnection.ConnectionString = this._sourceConnectionString;
                srcConnection.Open();
                //读取源表的数据
                if (this.SrcDbType == DatabaseType.ODBC)
                    adapter = new OdbcDataAdapter(_setting.LoadCommand, (OdbcConnection)srcConnection);
                else if (SrcDbType == DatabaseType.MySQL)
                    adapter = new MySqlDataAdapter(_setting.LoadCommand, (MySqlConnection)srcConnection);
                else if (this.SrcDbType == DatabaseType.MSSQL)
                    adapter = new SqlDataAdapter(_setting.LoadCommand, (SqlConnection)srcConnection);
                adapter.Fill(ds);

                string fileName = String.Empty;
                //将Datatable中的数据写入到文件中
                if (_setting.MediaType == MediaType.XmlFile)
                    DataTableToFile(ds.Tables[0], _setting.FilePath);
                else if (_setting.MediaType == MediaType.NormalFile)
                    DataTableToFile(ds.Tables[0], _setting.FilePath, ref fileName);
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ds.Clear();
                ds.Dispose();
                srcConnection.Close();
                srcConnection.Dispose();
            }
        }

        /// <summary>
        /// 将源数据库中的数据导出成指定的文件格式[单线程下限探测读取法]
        /// </summary>
        /// <param name="readOneTime"></param>
        protected virtual void ExportDateToFileByLowLimit()
        {
            IDbConnection srcConnection = null;
            IDbCommand srcCommand = null;
            IDataAdapter adapter = null;

            //源数据库类型连接
            if (this.SrcDbType == DatabaseType.ODBC)
            {
                srcConnection = new OdbcConnection();
                srcCommand = new OdbcCommand();
            }
            else if (this.SrcDbType == DatabaseType.MySQL)
            {
                srcConnection = new MySqlConnection();
                srcCommand = new MySqlCommand();
            }
            else if (this.SrcDbType == DatabaseType.MSSQL)
                throw new NotSupportedException("设置的方法不支持MSSQL");

            DataSet ds = new DataSet();
            int beginCount = 0;

            try
            {
                srcConnection.ConnectionString = this._sourceConnectionString;
                srcConnection.Open();
                //读取源表的数据
                if (this.SrcDbType == DatabaseType.ODBC)
                    adapter = new OdbcDataAdapter((OdbcCommand)srcCommand);
                else if (SrcDbType == DatabaseType.MySQL)
                    adapter = new MySqlDataAdapter((MySqlCommand)srcCommand);
                else if (this.SrcDbType == DatabaseType.MSSQL)
                    throw new NotSupportedException("设置的方法不支持MSSQL");

                srcCommand.Connection = srcConnection;
                string fileName = String.Empty;
                while(true)
                {
                    srcCommand.CommandText = String.Format(this._setting.LoadCommand, beginCount, this._setting.NumberOfReadOnce);
                    if (srcCommand.Connection.State != ConnectionState.Open)
                        srcCommand.Connection.Open();
                    adapter.Fill(ds);
                    
                    //将Datatable中的数据写入到文件中
                    if (_setting.MediaType == MediaType.XmlFile)
                        DataTableToFile(ds.Tables[0], _setting.FilePath);
                    else if (_setting.MediaType == MediaType.NormalFile)
                        DataTableToFile(ds.Tables[0], _setting.FilePath, ref fileName);

                    if (ds.Tables[0].Rows.Count != this._setting.NumberOfReadOnce)
                        break;
                    else
                    {
                        beginCount += this._setting.NumberOfReadOnce;
                        srcCommand.Connection.Close();
                        ds.Clear();
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ds.Clear();
                ds.Dispose();
                srcConnection.Close();
                srcConnection.Dispose();
            }
        }

        /// <summary>
        /// 将源数据库中的数据导出成指定的文件格式[范围数据][单线程下限探测读取法]
        /// </summary>
        protected virtual void ExportRangeDataToFileByLowLimit()
        {
            IDbConnection srcConnection = null;
            IDbCommand srcCommand = null;
            IDataAdapter adapter = null;

            //源数据库类型连接
            if (this.SrcDbType == DatabaseType.ODBC)
            {
                srcConnection = new OdbcConnection();
                srcCommand = new OdbcCommand();
            }
            else if (this.SrcDbType == DatabaseType.MySQL)
            {
                srcConnection = new MySqlConnection();
                srcCommand = new MySqlCommand();
            }
            else if (this.SrcDbType == DatabaseType.MSSQL)
                throw new NotSupportedException("设置的方法不支持MSSQL");

            DataSet ds = new DataSet();
            int beginCount = 0;

            try
            {
                srcConnection.ConnectionString = this._sourceConnectionString;
                srcConnection.Open();
                //读取源表的数据
                if (this.SrcDbType == DatabaseType.ODBC)
                    adapter = new OdbcDataAdapter((OdbcCommand)srcCommand);
                else if (SrcDbType == DatabaseType.MySQL)
                    adapter = new MySqlDataAdapter((MySqlCommand)srcCommand);
                else if (this.SrcDbType == DatabaseType.MSSQL)
                    throw new NotSupportedException("设置的方法不支持MSSQL");

                srcCommand.Connection = srcConnection;
                string fileName = String.Empty;
                int minValue = this._setting.MinValue;
                int tempMinValue = this._setting.MinValue;

                while (minValue != this._setting.MaxValue)
                {
                    while (true)
                    {
                    REFILLDS:
                        try
                        {
                            srcCommand.CommandText = String.Format(this._setting.LoadCommand, minValue, beginCount, this._setting.NumberOfReadOnce);
                            if (srcCommand.Connection.State != ConnectionState.Open)
                                srcCommand.Connection.Open();
                            adapter.Fill(ds);
                        }
                        catch
                        {
                            ds.Clear();
                            srcCommand.CommandText = "";
                            if(srcCommand.Connection.State != ConnectionState.Closed)
                                srcCommand.Connection.Close();
                            goto REFILLDS;
                        }
                        //将Datatable中的数据写入到文件中
                        //如果数据集中不存在记录,则不需要生成文件
                        if (ds.Tables[0].Rows.Count != 0)
                        {
                            Console.WriteLine(String.Format(this._setting.LoadCommand, minValue, beginCount, this._setting.NumberOfReadOnce));
                            if (_setting.MediaType == MediaType.XmlFile)
                                DataTableToFile(ds.Tables[0], _setting.FilePath);
                            else if (_setting.MediaType == MediaType.NormalFile)
                                DataTableToFile(ds.Tables[0], _setting.FilePath, ref fileName);
                        }
                        if (ds.Tables[0].Rows.Count != this._setting.NumberOfReadOnce)
                        {
                            ds.Clear();
                            break;
                        }
                        else
                        {
                            beginCount += this._setting.NumberOfReadOnce;
                            srcCommand.Connection.Close();
                            ds.Clear();
                        }
                    }
                    //如果存在二级步长
                    if (this._setting.SubStepValue != -1)
                    {
                        if (minValue < tempMinValue + this._setting.SubMaxValue)
                            minValue = minValue + this._setting.SubStepValue;
                        else
                        {
                            minValue = minValue - this._setting.SubMaxValue;
                            minValue = minValue + this._setting.StepValue;
                            tempMinValue = minValue;
                        }
                    }
                    else
                        minValue = minValue + this._setting.StepValue;
                    beginCount = 0;
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                ds.Clear();
                ds.Dispose();
                srcConnection.Close();
                srcConnection.Dispose();
            }
        }

        Thread[] threads;
        AutoResetEvent[] resetEvent;
        /// <summary>
        /// 将源数据库中的数据导出成指定的文件格式[范围数据][多线程]
        /// </summary>
        protected virtual void ExportRangeDataToFile(int threadNumber)
        {
            if (threadNumber == -1) return;

            //读取源表中数据
            IDbConnection srcConnection = null;
            IDbCommand command = null;
            IDataAdapter adapter = null;
            //源数据库类型连接
            if (this.SrcDbType == DatabaseType.MySQL)
            {
                srcConnection = new MySqlConnection();
                command = new MySqlCommand();
                adapter = new MySqlDataAdapter((MySqlCommand)command);
            }
            else if (this.SrcDbType == DatabaseType.MSSQL)
            {
                throw new NotSupportedException("当前版本类库不支持指定类型的数据库");
            }
            else if (this.SrcDbType == DatabaseType.ODBC)
            {
                srcConnection = new OdbcConnection();
                command = new OdbcCommand();
                adapter = new OdbcDataAdapter((OdbcCommand)command);
            }
            DataSet recCountSet = new DataSet();
            DataSet ds = new DataSet();

            try
            {
                //获取类型数量
                srcConnection.ConnectionString = this._sourceConnectionString;
                srcConnection.Open();
                command.Connection = srcConnection;
                command.CommandText = _setting.GetCountCommand;
                adapter.Fill(recCountSet);
                srcConnection.Close();
                if (recCountSet.Tables.Count == 0)
                    return;
                if(recCountSet.Tables[0].Rows.Count == 0)
                    return;
                if(!(recCountSet.Tables[0].Rows[0][1] is long))
                //循环获取得到的表进行分段
                    throw new ArgumentException("准备语句的第二个字段域必须是数字");

                int oneThreadRecCount = 0;
                //循环所有字段域
                foreach (DataRow recRow in recCountSet.Tables[0].Rows)
                {
                    //如果当前字段域的值小于一次获取数据的量,使用当前线程一次性读取
                    if (Convert.ToInt32(recRow[1]) <= this._setting.NumberOfReadOnce)
                    {
                        Console.WriteLine("线程" + Thread.CurrentThread.Name + "开始导出" + _setting.Id + "数据,数据量:" + recRow[1] + " Key值为" + recRow[0]);
                        command.CommandText = String.Format(this._setting.LoadCommand, recRow[0], 0, this._setting.NumberOfReadOnce);
                        if (command.Connection.State != ConnectionState.Open)
                            command.Connection.Open();
                        adapter.Fill(ds);
                        DataTableToFile(ds.Tables[0], this._setting.FilePath);
                        command.Connection.Close();
                        ds.Clear();
                    }
                    //如果当前字段域的值大于一次获取数据的量,使用多线程进行读取
                    else
                    {
                        //进行数据分段
                        Console.WriteLine("线程" + Thread.CurrentThread.Name + "开始导出" + _setting.Id + "数据,数据量:" + recRow[1] + " Key值为" + recRow[0]);
                        oneThreadRecCount = (int)Math.Floor(Convert.ToDecimal(recRow[1]) / threadNumber);
                        threads = new Thread[threadNumber];
                        resetEvent = new AutoResetEvent[threadNumber];
                        string args = String.Empty;
                        //开启线程
                        for (int i = 0; i < threadNumber; i++)
                        {
                            //参数序列 线程ID|开始记录数|结束记录数
                            if ((i + 1) != threadNumber)
                                args = String.Format("{0}|{1}|{2}|{3}", i, i * oneThreadRecCount, (i + 1) * oneThreadRecCount, recRow[0]);
                            else
                                args = String.Format("{0}|{1}|{2}|{3}", i, i * oneThreadRecCount, recRow[1], recRow[0]);
                            //启动线程
                            resetEvent[i] = new AutoResetEvent(false);
                            threads[i] = new Thread(new ParameterizedThreadStart(ExportToFileInThread));
                            threads[i].IsBackground = true;
                            threads[i].Name = "Thread_" + i;
                            threads[i].Start(args);
                        }
                        //主线程等待
                        WaitHandle.WaitAll(resetEvent);
                        //清理现场
                        for (int innerCount = 0; innerCount < threadNumber; innerCount++)
                        {
                            threads[innerCount] = null;
                            resetEvent[innerCount].Close();
                            resetEvent[innerCount] = null;
                        }
                    }
                }
            }
            catch
            {
                throw;
            }
        }

        /// <summary>
        /// 将源数据库中的数据导出成指定的文件格式[多线程]
        /// </summary>
        /// <param name="threadNumber">线程数量</param>
        protected virtual void ExportDataToFile(int threadNumber)
        {
            if (threadNumber == -1) return;

            //读取源表中数据
            IDbConnection srcConnection = null;
            IDbCommand command = null;
            //源数据库类型连接
            if (this.SrcDbType == DatabaseType.MySQL)
            {
                srcConnection = new MySqlConnection();
                command = new MySqlCommand();
            }
            else if (this.SrcDbType == DatabaseType.MSSQL)
            {
                srcConnection = new SqlConnection();
                command = new SqlCommand();
            }
            else if (this.SrcDbType == DatabaseType.ODBC)
            {
                srcConnection = new OdbcConnection();
                command = new OdbcCommand();
            }

            try
            {
                srcConnection.ConnectionString = this._sourceConnectionString;
                srcConnection.Open();
                command.Connection = srcConnection;
                command.CommandText = _setting.GetCountCommand;
                int recCount = -1;
                recCount = Convert.ToInt32(command.ExecuteScalar());
                srcConnection.Close();
                if (recCount == 0)
                    return;

                //开始分段
                Console.WriteLine("线程" + Thread.CurrentThread.Name + "开始导出" + _setting.Id + "数据,数据量:" + recCount);
                int oneThreadRecCount = (int)Math.Floor((double)recCount / threadNumber);
                threads = new Thread[threadNumber];
                resetEvent = new AutoResetEvent[threadNumber];
                string args = String.Empty;

                //开启线程
                for (int i = 0; i < threadNumber; i++)
                {
                    //参数序列 线程ID|开始记录数|结束记录数
                    if ((i + 1) != threadNumber)
                        args = String.Format("{0}|{1}|{2}", i, i * oneThreadRecCount, (i + 1) * oneThreadRecCount);
                    else
                        args = String.Format("{0}|{1}|{2}", i, i * oneThreadRecCount, recCount);
                    //启动线程
                    resetEvent[i] = new AutoResetEvent(false);
                    threads[i] = new Thread(new ParameterizedThreadStart(ExportToFileInThread));
                    threads[i].IsBackground = true;
                    threads[i].Name = "Thread_" + i;
                    threads[i].Start(args);
                }
                //主线程等待
                WaitHandle.WaitAll(resetEvent);
                //清理现场
                for (int innerCount = 0; innerCount < threadNumber; innerCount++)
                {
                    threads[innerCount] = null;
                    resetEvent[innerCount].Close();
                    resetEvent[innerCount] = null;
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                if (srcConnection.State != ConnectionState.Closed)
                    srcConnection.Close();
                srcConnection.Dispose();
                command.Dispose();
            }
        }

        /// <summary>
        /// 将导入的文件导入到目标表中
        /// </summary>
        protected virtual void TransportFileToTargeDatabase()
        {
            //if (this._tagDbType == DatabaseType.MySQL)
            //    return;

            int retryTime = 0;

            SqlConnection connection = new SqlConnection();
            SqlCommand command = new SqlCommand();
            try
            {
                connection.ConnectionString = this._targetConnectionString;
                connection.Open();

                command.Connection = connection;
                command.CommandTimeout = connection.ConnectionTimeout;

                if (this._setting.TransportType == TransportType.ReLoad)
                {
                    if (this._tagDbType == DatabaseType.MSSQL)
                    {
                        command.CommandText = "Truncate Table " + this._setting.TargetTable;
                        command.ExecuteNonQuery();
                    }
                }

                string[] files = Directory.GetFiles(this._setting.FilePath);
                foreach (string file in files)
                {
                RETRY:
                    try
                    {
                        if (this.OnFileImport != null)
                            this.OnFileImport(file);
                        else
                        {
                            if (this._setting.MediaType != MediaType.Database)
                                command.CommandText = String.Format(this._setting.InsertCommand, file, this._setting.TargetTable);
                            else
                                command.CommandText = String.Format(this._setting.InsertCommand, this._setting.Id, file);
                            if (command.Connection.State != ConnectionState.Open)
                                command.Connection.Open();
                            Console.WriteLine("[" + this._setting.Id + "]:导入" + file + "成功~");
                            command.ExecuteNonQuery();
                        }
                        File.Delete(file);
                        //当成功执行语句后,如果曾经重试过,归零
                        if (retryTime != 0)
                            retryTime = 0;
                    }
                    catch (Exception ex)
                    {
                        if (retryTime == 5)
                        {
                            retryTime = 0;
                            Console.WriteLine("多次尝试后,仍然导入失败");
                            continue;
                        }
                        else
                        {
                            Console.WriteLine("当前线程[" + Thread.CurrentThread.Name + "]" + ex.Message);
                            Console.WriteLine("导入命令=" + command.CommandText);
                            Thread.Sleep(new TimeSpan(0, 0, 0, 30, 0)); //线程等待30秒
                            retryTime++;
                            goto RETRY;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("当前线程[" + Thread.CurrentThread.Name + "]" + ex.Message);
                throw ex;
            }
            finally
            {
                if (connection.State != ConnectionState.Closed)
                    connection.Close();
                connection.Dispose();
                command.Dispose();
            }
        }

        /// <summary>
        /// 将临时表中的数据导入到正式表中
        /// </summary>
        protected virtual void TransportTempData()
        {
            SqlConnection connection = new SqlConnection();
            SqlCommand command = new SqlCommand();
            SqlCommand command2 = new SqlCommand();
            try
            {
                connection.ConnectionString = this._targetConnectionString;
                connection.Open();

                command.Connection = connection;
                command.CommandTimeout = connection.ConnectionTimeout;
                command.CommandText = "Select AutoID From TransportXML Where SectionID='" + this._setting.Id + "'";
                SqlDataReader reader = command.ExecuteReader(CommandBehavior.CloseConnection);

                while (reader.Read())
                {
                    command2.CommandText = String.Format(this._setting.ExportCommand, reader["AutoID"], this._setting.TargetTable);
                    if (command2.Connection.State != ConnectionState.Open)
                        command2.Connection.Open();
                    command2.ExecuteNonQuery();
                    Console.WriteLine("[" + this._setting.Id + "]:成功导入临时中转表" + reader["AutoID"].ToString() + "号记录");
                }
                reader.Close();
                //删除记录
                command.CommandText = "Delete TransportXML Where SectionID='" + this._setting.Id + "'";
                if (command.Connection.State != ConnectionState.Open)
                    command.Connection.Open();
                command.ExecuteNonQuery();
            }
            catch (Exception ex)
            {
                Console.WriteLine("当前线程[" + Thread.CurrentThread.Name + "]" + ex.Message);
                throw ex;
            }
            finally
            {
                if (connection.State != ConnectionState.Closed)
                    connection.Close();
                connection.Dispose();
                command.Dispose();
                command2.Dispose();
            }
        }
        /// <summary>
        /// 检查执行周期
        /// </summary>
        /// <returns>是否到达执行周期</returns>
        private bool CheckFrequency()
        {
            //如果没有设置执行周期控制表
            if (String.IsNullOrEmpty(this._frequencyTable))
                return true; //总是要执行

            if (this._tagDbType == DatabaseType.FTPFile)
                return true;    

            IDbConnection connection = null;
            IDbCommand command = null;
            IDataReader reader = null;

            if (this._tagDbType == DatabaseType.MSSQL)
            {
                connection = new SqlConnection();
                command = new SqlCommand();
            }
            else if (this._tagDbType == DatabaseType.MySQL)
            {
                connection = new MySqlConnection();
                command = new MySqlCommand();
            }
            else if (this._tagDbType == DatabaseType.ODBC)
            {
                connection = new OdbcConnection();
                command = new OdbcCommand();
            }

            connection.ConnectionString = this._targetConnectionString;
            int frequency = 1;
            int frequencyUnit = 1; //1=以天为单位 2=以小时为单位 3=以分钟为单位 4=以月为单位 5=以年为单位
            DateTime lastExecute = DateTime.Now;

            //获取设置中指定ID的执行频度
            try
            {
                connection.Open();
                command.Connection = connection;
                command.CommandText = "Select * From " + this._frequencyTable + " Where TransportID='" + this._setting.Id + "'";
                reader = command.ExecuteReader(CommandBehavior.CloseConnection);

                //如果存在这样的记录
                if (reader.Read())
                {
                    Int32.TryParse(reader[1].ToString(), out frequency);
                    Int32.TryParse(reader[2].ToString(), out frequencyUnit);
                    DateTime.TryParse(reader[3].ToString(), out lastExecute);

                    //执行周期为天
                    if (frequencyUnit == 1)
                    {
                        if (lastExecute.AddDays((double)frequency) >= DateTime.Today)
                            return true;
                        else
                            return false;
                    }
                    //执行周期为小时
                    else if (frequencyUnit == 2)
                    {
                        if (lastExecute.AddHours((double)frequency) >= DateTime.Now)
                            return true;
                        else
                            return false;
                    }
                    //执行周期为分钟
                    else if (frequencyUnit == 3)
                    {
                        if (lastExecute.AddMinutes((double)frequency) >= DateTime.Now)
                            return true;
                        else
                            return false;
                    }
                    //执行周期为月
                    else if (frequencyUnit == 4)
                    {
                        if (lastExecute.AddMonths(frequency) >= DateTime.Today)
                            return true;
                        else
                            return false;
                    }
                    //执行周期为年
                    else if (frequencyUnit == 5)
                    {
                        if (lastExecute.AddYears(frequency) >= DateTime.Today)
                            return true;
                        else
                            return false;
                    }
                    else
                        return false;
                }
                //如果不存在,就去执行
                else
                    return true;
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                if (connection.State != ConnectionState.Closed)
                    connection.Close();
                connection.Dispose();
                command.Dispose();
            }
        }

        /// <summary>
        /// 执行成功后,更新执行周期
        /// </summary>
        private void UpdateFrequency()
        {
            //如果没有设置执行周期控制表
            if (String.IsNullOrEmpty(this._frequencyTable))
                return; //总是要执行

            if (this._tagDbType == DatabaseType.FTPFile)
                return;

            IDbConnection connection = null;
            IDbCommand command = null;
            IDataReader reader = null;

            if (this._tagDbType == DatabaseType.MSSQL)
            {
                connection = new SqlConnection();
                command = new SqlCommand();
            }
            else if (this._tagDbType == DatabaseType.MySQL)
            {
                connection = new MySqlConnection();
                command = new MySqlCommand();
            }
            else if (this._tagDbType == DatabaseType.ODBC)
            {
                connection = new OdbcConnection();
                command = new OdbcCommand();
            }

            connection.ConnectionString = this._targetConnectionString;
            int frequency = 1;
            int frequencyUnit = 1; //1=以天为单位 2=以小时为单位 3=以分钟为单位 4=以月为单位 5=以年为单位
            DateTime lastExecute = DateTime.Now;

            //获取设置中指定ID的执行频度
            try
            {
                connection.Open();
                command.Connection = connection;
                command.CommandText = "Select * From " + this._frequencyTable + " Where TransportID='" + this._setting.Id + "'";
                reader = command.ExecuteReader(CommandBehavior.CloseConnection);

                //如果存在这样的记录
                if (reader.Read())
                {
                    Int32.TryParse(reader[1].ToString(), out frequency);
                    Int32.TryParse(reader[2].ToString(), out frequencyUnit);
                    DateTime.TryParse(reader[3].ToString(), out lastExecute);

                    //执行周期为天
                    //执行周期为月
                    //执行周期为年
                    if (frequencyUnit == 1 || frequencyUnit == 4 || frequencyUnit == 5)
                        command.CommandText = "Update " + this._frequencyTable + " Set LastExecute='" + DateTime.Today + "' Where TransportID='" + this._setting.Id + "'";
                    //执行周期为小时
                    else if (frequencyUnit == 2)
                        command.CommandText = "Update " + this._frequencyTable + " Set LastExecute='" + DateTime.Now.ToString("yyyy-MM-dd hh:00:00") + "' Where TransportID='" + this._setting.Id + "'";
                    //执行周期为分钟
                    else if (frequencyUnit == 3)
                        command.CommandText = "Update " + this._frequencyTable + " Set LastExecute='" + DateTime.Now.ToString("yyyy-MM-dd hh:mm:00") + "' Where TransportID='" + this._setting.Id + "'";

                    if (command.Connection.State != ConnectionState.Open)
                        command.Connection.Open();
                    command.ExecuteNonQuery();
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                if (connection.State != ConnectionState.Closed)
                    connection.Close();
                connection.Dispose();
                command.Dispose();
            }
        }

        /// <summary>
        /// 导入文件的线程执行方法
        /// </summary>
        /// <param name="args">线程参数</param>
        private void ExportToFileInThread(object threadArgs)
        {
            string[] args = threadArgs.ToString().Split('|');
            int threadID = Convert.ToInt32(args[0]);
            int startRec = Convert.ToInt32(args[1]), endRec = Convert.ToInt32(args[2]); //开始记录号,结束记录号
            string arg4 = String.Empty;
            if (args.Length == 4)
                arg4 = args[3];
            string fileName = String.Empty; //文件名
            int counter = 0;

            if (this._setting.LoadCommand == String.Empty)
            {
                resetEvent[threadID].Set();
                return;
            }
            //得到线程要处理的数据量
            int recCount = endRec - startRec;
            //声明服务器连接
            IDbConnection srcConnection = null;
            IDataAdapter adapter = null;
            IDbCommand selectCommand = null;
            DataSet ds = new DataSet();

            //源数据库类型连接
            if (this.SrcDbType == DatabaseType.ODBC)
                srcConnection = new OdbcConnection();
            else if (this.SrcDbType == DatabaseType.MySQL)
                srcConnection = new MySqlConnection();
            else if (this.SrcDbType == DatabaseType.MSSQL)
                srcConnection = new SqlConnection();

            //开始读取
            try
            {
                srcConnection.ConnectionString = this._sourceConnectionString;
                srcConnection.Open();

                if (this.SrcDbType == DatabaseType.MySQL)
                {
                    adapter = new MySqlDataAdapter();
                    selectCommand = new MySqlCommand();
                    selectCommand.Connection = srcConnection;
                }
                else if (this.SrcDbType == DatabaseType.ODBC)
                {
                    adapter = new OdbcDataAdapter();
                    selectCommand = new OdbcCommand();
                    selectCommand.Connection = srcConnection;
                }
                else if (this.SrcDbType == DatabaseType.MSSQL)
                {
                    adapter = new SqlDataAdapter();
                    selectCommand = new SqlCommand();
                    selectCommand.Connection = srcConnection;
                }

                string result = String.Empty;
                //开始读取源数据
                for (counter = 0; counter < (recCount / this._setting.NumberOfReadOnce) + 1; counter++)
                {
                RETRYEXECUTE:
                    if (counter < (recCount / this._setting.NumberOfReadOnce))
                    {
                        if(arg4 == String.Empty)
                            selectCommand.CommandText = String.Format(this._setting.LoadCommand, 
                                (counter * this._setting.NumberOfReadOnce) + startRec, this._setting.NumberOfReadOnce);
                        else
                            selectCommand.CommandText = String.Format(this._setting.LoadCommand, arg4,
                                (counter * this._setting.NumberOfReadOnce) + startRec, this._setting.NumberOfReadOnce);
                    }
                    else
                    {
                        if(arg4 == String.Empty)
                            selectCommand.CommandText = String.Format(this._setting.LoadCommand, 
                                (counter * this._setting.NumberOfReadOnce) + startRec, endRec - ((counter * this._setting.NumberOfReadOnce) + startRec));
                        else
                            selectCommand.CommandText = String.Format(this._setting.LoadCommand, arg4,
                                (counter * this._setting.NumberOfReadOnce) + startRec, endRec - ((counter * this._setting.NumberOfReadOnce) + startRec));
                    }
                    try
                    {
                        if (selectCommand.Connection.State != ConnectionState.Open)
                            selectCommand.Connection.Open();
                        if (this.SrcDbType == DatabaseType.MySQL)
                        {
                            ((MySqlDataAdapter)adapter).SelectCommand = (MySqlCommand)selectCommand;
                            adapter.Fill(ds);
                            ((MySqlDataAdapter)adapter).SelectCommand.Connection.Close();
                        }
                        else if (this.SrcDbType == DatabaseType.ODBC)
                        {
                            ((OdbcDataAdapter)adapter).SelectCommand = (OdbcCommand)selectCommand;
                            adapter.Fill(ds);
                            ((OdbcDataAdapter)adapter).SelectCommand.Connection.Close();
                        }
                        else if (this.SrcDbType == DatabaseType.MSSQL)
                        {
                            ((SqlDataAdapter)adapter).SelectCommand = (SqlCommand)selectCommand;
                            adapter.Fill(ds);
                            ((SqlDataAdapter)adapter).SelectCommand.Connection.Close();
                        }
                        //不同的文件传输方式
                        if (this._setting.MediaType == MediaType.NormalFile)
                            DataTableToFile(ds.Tables[0], this._setting.FilePath, ref fileName);
                        else if (this._setting.MediaType == MediaType.XmlFile)
                            DataTableToFile(ds.Tables[0], this._setting.FilePath);
                        ds.Clear();
                    }
                    catch(Exception ex)
                    {
                        Console.WriteLine("当前线程[" + Thread.CurrentThread.Name + "][" + _setting.Id + "]" + ex.Message);
                        Console.WriteLine("出错语句" + selectCommand.CommandText);
                        //如果程序执行过程中出现连接丢失的现象,者将线程休息1分钟后,再尝试连接
                        Thread.Sleep(new TimeSpan(0, 1, 0));
                        if (srcConnection.State != ConnectionState.Closed)
                            srcConnection.Close();

                        srcConnection.Dispose();
                        srcConnection = null;
                        selectCommand.Dispose();
                        selectCommand = null;

                        //源数据库类型连接
                        if (this.SrcDbType == DatabaseType.ODBC)
                        {
                            srcConnection = new OdbcConnection();
                            selectCommand = new OdbcCommand();
                        }
                        else if (this.SrcDbType == DatabaseType.MySQL)
                        {
                            srcConnection = new MySqlConnection();
                            selectCommand = new MySqlCommand();
                        }
                        else if (this.SrcDbType == DatabaseType.MSSQL)
                        {
                            srcConnection = new SqlConnection();
                            selectCommand = new SqlCommand();
                        }

                        srcConnection.ConnectionString = this._sourceConnectionString;
                        srcConnection.Open();
                        selectCommand.Connection = srcConnection;

                        goto RETRYEXECUTE;
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                if (srcConnection.State != ConnectionState.Closed)
                    srcConnection.Close();

                srcConnection.Dispose();
                srcConnection = null;
                selectCommand.Dispose();
                selectCommand = null;
                resetEvent[threadID].Set();
            }

        }

        /// <summary>
        /// 将数据表的数据写入到文件中
        /// </summary>
        /// <param name="table"></param>
        private void DataTableToFile(DataTable table, string path)
        {
            if (String.IsNullOrEmpty(path)) throw new ArgumentNullException("目标路径为空");
            if (!Directory.Exists(path))
                Directory.CreateDirectory(path);

            string fileName = Guid.NewGuid().ToString();
            table.WriteXml(path + @"/" + fileName + ".xml");
            Console.WriteLine("线程[" + Thread.CurrentThread.Name + "][" + this._setting.Id + "]:写入文件" + fileName + ";");
        }

        /// <summary>
        /// 将数据记录写入文件中
        /// </summary>
        /// <param name="row">导入的行</param>
        /// <param name="fileName">文件名</param>
        private void DataTableToFile(DataTable table, string path, ref string fileName)
        {
            if (String.IsNullOrEmpty(path)) throw new ArgumentNullException("目标路径为空");
            if (!Directory.Exists(path))
                Directory.CreateDirectory(path);

            //初始化
            if (fileName == String.Empty)
            {
                fileName = Guid.NewGuid().ToString();
                //建立这个文件
                FileStream fs = new FileStream(path + @"\" + fileName, FileMode.CreateNew, FileAccess.ReadWrite);
                fs.Close();
                fs.Dispose();
            }
            //如果是一个已经存在的文件
            else
            {
                FileInfo fi = new FileInfo(path + @"\" + fileName);
                //文件存在
                if (fi.Exists)
                {
                    //如果这个文件的大小大于5M
                    if (fi.Length >= (5 * 1024 * 1024))
                    {
                        //建立一个新的文件
                        fileName = Guid.NewGuid().ToString();
                        //建立这个文件
                        FileStream fs = new FileStream(path + @"\" + fileName, FileMode.CreateNew, FileAccess.ReadWrite);
                        fs.Close();
                        fs.Dispose();
                    }
                }
                //文件不存在
                else
                {
                    //建立一个新的文件
                    fileName = Guid.NewGuid().ToString();
                    //建立这个文件
                    FileStream fs = new FileStream(path + @"\" + fileName, FileMode.CreateNew, FileAccess.ReadWrite);
                    fs.Close();
                    fs.Dispose();
                }
            }
            string result = String.Empty;
            using (StreamWriter sw = new StreamWriter(path + @"\" + fileName, true))
            {
                foreach (DataRow row in table.Rows)
                {
                    //如果用户自定义了文件写入的模式
                    if (this.OnDataRowWriting != null)
                        result = this.OnDataRowWriting(row);
                    //连接所有的字段
                    else
                    {
                        foreach (DataColumn column in table.Columns)
                            result += (row[column] + ",");
                        result = result.TrimEnd(',');
                    }
                    sw.WriteLine(result);
                    result = String.Empty;
                }
            }
        }

        #region IDisposable Members

        public void Dispose()
        {
            
        }

        #endregion
    }

    /// <summary>
    /// 操作传输综合类
    /// </summary>
    public class OperateTransmission : TransmissionBase, IDisposable
    {
        private string _sourceConnectionString = String.Empty;
        private string _targetConnectionString = String.Empty;
        private DatabaseType _srcDbType = DatabaseType.MySQL;
        private DatabaseType _tagDbType = DatabaseType.MSSQL;
        private OperateSetting _setting = null;
        private Logger logger;
        private static string _LogPath = AppDomain.CurrentDomain.BaseDirectory + @"\Log\DataOperate_" + DateTime.Today.ToString("yyyyMMdd") + ".log";

        public override DatabaseType SrcDbType
        {
            get { return _srcDbType; }
            set { _srcDbType = value; }
        }

        public override string SourceConnectionString
        {
            get { return _sourceConnectionString; }
            set { _sourceConnectionString = value; }
        }

        public override DatabaseType TagDbType
        {
            get { return _tagDbType; }
            set { _tagDbType = value; }
        }

        public override string TargetConnectionString
        {
            get { return _targetConnectionString; }
            set { _targetConnectionString = value; }
        }

        /// <summary>
        /// 数据库操作设置
        /// </summary>
        public OperateSetting Setting
        {
            get { return _setting; }
            set { _setting = value; }
        }

        #region 构造函数
        public OperateTransmission(OperateSetting setting)
        {
            if(setting != null)
                this._setting = setting;
            logger = Logger.CreateLogerInstance();
            logger.LogPath = _LogPath;
        }
        public OperateTransmission() : this(null)
        {

        }
        #endregion

        /// <summary>
        /// 开始执行配置的操作
        /// </summary>
        public override void Begin()
        {
            if (this._setting.OperateType == OperateType.CreateIndex)
            {
                if (this._setting.TargetDatabase == TargetDatabase.Source)
                    CreateIndex(_srcDbType, this._setting.TargetDatabase);
                else
                    CreateIndex(_tagDbType, this._setting.TargetDatabase);
            }
            else
            {
                throw new NotImplementedException();
            }
        }

        protected virtual void CreateIndex(DatabaseType dbType, TargetDatabase target)
        {
            IDbCommand command = null;
            IDbDataAdapter adapter = null;
            IDbConnection connection = null;
            string connString = String.Empty;

            if (target == TargetDatabase.Source)
                connString = this._sourceConnectionString;
            else
                connString = this._targetConnectionString;

            DataSet ds = new DataSet();
            //组合建立索引需要的列
            string columns = String.Empty;
            foreach(string column in this._setting.IndexAttributes.IndexColumns)
                columns += column + ",";
            columns = columns.TrimEnd(',');

            try
            {
                switch (dbType)
                {
                    case DatabaseType.ODBC:
                        adapter = new OdbcDataAdapter();
                        connection = new OdbcConnection(connString);
                        command = new OdbcCommand();
                        command.Connection = (OdbcConnection)connection;
                        adapter.SelectCommand = (OdbcCommand)command;
                        break;
                    case DatabaseType.MySQL:
                        adapter = new MySqlDataAdapter();
                        connection = new MySqlConnection(connString);
                        command = new MySqlCommand();
                        command.Connection = (MySqlConnection)connection;
                        adapter.SelectCommand = (MySqlCommand)command;
                        break;
                    case DatabaseType.MSSQL:
                        connection = new SqlConnection(connString);
                        command = new SqlCommand();
                        command.Connection = (SqlConnection)connection;
                        if (command.Connection.State != ConnectionState.Open)
                            command.Connection.Open();
                        if (this._setting.IndexAttributes.IsClustered)
                            command.CommandText = String.Format("IF EXISTS(Select name From sys.indexes Where name=N'{0}) Drop Index {0} On {1};GO Create Unique Index {0} On {1}({2})",
                                this._setting.ObjectName, this._setting.IndexAttributes.TableName, columns);
                        else
                            command.CommandText = String.Format("IF EXISTS(Select name From sys.indexes Where name=N'{0}) Drop Index {0} On {1};GO Create Index {0} On {1}({2})",
                                this._setting.ObjectName, this._setting.IndexAttributes.TableName, columns);
                        command.ExecuteNonQuery();
                        break;
                    case DatabaseType.FTPFile:
                        throw new NotSupportedException("不支持的数据类型");
                }
                if (dbType == DatabaseType.ODBC || dbType == DatabaseType.MySQL)
                {
                    command.CommandText = "Show Index From " + this._setting.IndexAttributes.TableName + ";";
                    adapter.Fill(ds);
                    if (ds.Tables.Count == 0) throw new NullReferenceException("查询指定的表索引错误");
                    if (ds.Tables[0].Select("Column_name='" + columns + "'").Length == 0)
                    {
                        command.CommandText = String.Format("Create Index {0} On {1}({2})", this._setting.ObjectName, this._setting.IndexAttributes.TableName, columns);
                        if (command.Connection.State != ConnectionState.Open)
                            command.Connection.Open();
                        Console.WriteLine("正在建立索引" + this._setting.ObjectName + "....");
                        command.ExecuteNonQuery();
                        Console.WriteLine("完成");
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            finally
            {
                if (connection != null)
                {
                    if (connection.State != ConnectionState.Closed)
                        connection.Close();
                }
                connection.Dispose();
                if (command != null)
                    command.Dispose();
            }
        }

        protected virtual void DropIndex(DatabaseType dbType, TargetDatabase target)
        {
            throw new NotImplementedException("该方法尚未实现");
        }

        protected virtual void CreateTable(DatabaseType dbType, TargetDatabase target)
        {
            throw new NotImplementedException("该方法尚未实现");
        }

        protected virtual void DropTable(DatabaseType dbType, TargetDatabase target)
        {
            throw new NotImplementedException("该方法尚未实现");
        }

        #region IDisposable Members

        public void Dispose()
        {
            throw new NotImplementedException();
        }

        #endregion
    }

    public sealed class TransmissionFactory
    {
        private TransmissionFactory()
        {
        }

        /// <summary>
        /// 开始传输操作
        /// </summary>
        /// <param name="type">传输类型</param>
        /// <returns>返回对应的传输类</returns>
        public static TransmissionBase BeginTransmission(string type)
        {
            if (type == "operate")
                return new OperateTransmission();
            else if (type == "data")
                return new DataTransmission();
            else
                return null;
        }
    }
    
    /// <summary>
    /// 从FPT上下载文件的类
    /// </summary>
    public class FTPDownloader : IDisposable
    {
        private string _logFilePath = AppDomain.CurrentDomain.BaseDirectory + "FTPDownLoader.log";
        private Logger logger; //日志记录
        private int _retryTimes = 5; //下载重试次数
        private string _localPath = String.Empty; //本地存放路径
        private string[] _remoteFileFormat ; //远程文件格式
        private List<FTPEntity> _ftpList = new List<FTPEntity>();

        /// <summary>
        /// 下载重试次数
        /// </summary>
        public int RetryTimes
        {
            get { return _retryTimes; }
            set { _retryTimes = value; }
        }

        /// <summary>
        /// 本地存放路径
        /// </summary>
        public string LocalPath
        {
            get { return _localPath; }
            set { _localPath = value; }
        }

        /// <summary>
        /// 远程文件格式[复数]
        /// </summary>
        public string[] RemoteFileFormat
        {
            get { return _remoteFileFormat; }
            set { _remoteFileFormat = value; }
        }

        /// <summary>
        /// FTP地址列表
        /// </summary>
        public List<FTPEntity> FTPAddressList
        {
            get { return _ftpList; }
            set { _ftpList = value; }
        }

        #region 构造函数
        /// <summary>
        /// FTP下载器
        /// </summary>
        public FTPDownloader()
        {
            logger = Logger.CreateLogerInstance();
            logger.LogPath = _logFilePath;
        }

        ~FTPDownloader()
        {
            GC.SuppressFinalize(logger);
        }
        #endregion

        public void Download()
        {
            if (this._ftpList.Count == 0) throw new ArgumentNullException("下载地址列表不能为空");
            if (this._remoteFileFormat.Length == 0) throw new ArgumentNullException("远程文件名不能为空");

            ParameterFitter fitter = new ParameterFitter();
            //替换所有的远程文件名
            for (int i = 0; i < _remoteFileFormat.Length; i++)
                _remoteFileFormat[i] = fitter.Fit(_remoteFileFormat[i]);

            DownFTPClass objDown = new DownFTPClass();
            int result = 0;
            string message = String.Empty;

            List<FTPEntity> retryList = new List<FTPEntity>();
            List<FTPEntity> removeIndex = new List<FTPEntity>();
            int count = 0;

            Console.WriteLine("开始下载");
            logger.WriteLog("开始第一次下载文件");
            Console.WriteLine("开始向FTP服务器下载文件");

            string tempFileName = String.Empty;
            string tempFileName2 = String.Empty;

            foreach (FTPEntity entity in this._ftpList)
            {
                if (!Directory.Exists(this._localPath + @"\" + entity.ServerName))
                    Directory.CreateDirectory(this._localPath + @"\" + entity.ServerName);
                Console.Write((char)13 + "正在下载" + entity.ServerName + "的日志");

                //循环下载需要的文件
                foreach (string fileName in _remoteFileFormat)
                {
                    //使用自定义的路径
                    if (fileName.Contains("%dir%"))
                    {
                        tempFileName = fileName;
                        tempFileName = tempFileName.Remove(0, 5);
                        tempFileName2 = fileName.Substring(fileName.LastIndexOf("/") + 1);

                        objDown.BeginDownload("ftp://" + entity.ServerIP + tempFileName, 1, _localPath + @"\" + entity.ServerName + @"\" + tempFileName2, entity.UserName, entity.Password, out result, out message);
                    }
                    else
                        objDown.BeginDownload("ftp://" + entity.ServerIP + "/" + entity.SubDirectory + fileName, 1, _localPath + @"\" + entity.ServerName + @"\" + fileName, entity.UserName, entity.Password, out result, out message);

                    if (result != 0)
                    {
                        logger.WriteLog("下载" + fileName + "失败~" + message);
                        retryList.Add(entity);
                        continue;
                    }
                }
            }

            logger.WriteLog("第一次获取日志文件完成");
            int totalResult = 0; //是否所有文件都下载成功了
            for (int i = 0; i < _retryTimes; i++)
            {
                if (retryList.Count == 0) break;
                logger.WriteLog("开始第" + (i + 1) + "次尝试");
                Console.WriteLine();
                Console.WriteLine("开始第" + (i + 1) + "次尝试");
                count = retryList.Count;

                for (int j = 0; j < count; j++)
                {
                    Console.Write((char)13 + "正在下载" + retryList[j].ServerName + "的日志");
                    foreach (string fileName in _remoteFileFormat)
                    {
                        //如果文件已经被下载
                        if (File.Exists(_localPath + @"\" + retryList[j].ServerName + @"\" + fileName))
                            continue;
                        //下载
                        //使用自定义的路径
                        if (fileName.Contains("%dir%"))
                        {
                            tempFileName = fileName;
                            tempFileName = tempFileName.Remove(0, 5);
                            tempFileName2 = fileName.Substring(fileName.LastIndexOf("/") + 1);

                            objDown.BeginDownload("ftp://" + retryList[j].ServerIP + tempFileName, 1, _localPath + @"\" + retryList[j].ServerName + @"\" + tempFileName2, retryList[j].UserName, retryList[j].Password, out result, out message);
                        }
                        else
                            objDown.BeginDownload("ftp://" + retryList[j].ServerIP + "/" + retryList[j].SubDirectory + fileName, 1, _localPath + @"\" + retryList[j].ServerName + @"\" + fileName, retryList[j].UserName, retryList[j].Password, out result, out message);
                        totalResult += result;
                        if (result != 0)
                        {
                            logger.WriteLog("第" + (i + 1) + "次,下载" + fileName + "失败~" + message);
                            continue;
                        }
                    }

                    //如果所有的文件都下载成功
                    if(totalResult == 0)
                        removeIndex.Add(retryList[j]);
                    totalResult = 0;
                }
                //一次重试后,移除已经下载成功的部分
                foreach (FTPEntity r in removeIndex)
                    retryList.Remove(r);
                removeIndex.Clear();
            }

            //多次尝试后
            if (retryList.Count > 0)
            {
                message = String.Empty;
                foreach (FTPEntity ft in retryList)
                {
                    message += (ft.ServerName + ",");
                }
                logger.WriteLog("经过多次尝试,仍有未能下载的日志,程序将放弃这些日志,请手工处理:" + message);
            }
            logger.WriteLog("下载完成");
            Console.WriteLine("下载完成");
            retryList.Clear();
        }

        public void Download(List<FTPEntity> ftpList)
        {
            if (ftpList.Count == 0) throw new ArgumentNullException("ftpList");

            this._ftpList = ftpList;
            Download();
        }


        #region IDisposable Members

        public void Dispose()
        {
            this._ftpList.Clear();
        }

        #endregion
    }
    /// <summary>
    /// FTP地址实体
    /// </summary>
    public class FTPEntity
    {
        public FTPEntity()
        {
        }

        private int _serverID;
        /// <summary>
        /// 服务器ID
        /// </summary>
        public int ServerID
        {
            get { return _serverID; }
            set { _serverID = value; }
        }

        private string _serverName;
        /// <summary>
        /// 服务器名
        /// </summary>
        public string ServerName
        {
            get { return _serverName; }
            set { _serverName = value; }
        }

        private string _serverIP;
        /// <summary>
        /// 服务器IP
        /// </summary>
        public string ServerIP
        {
            get { return _serverIP; }
            set { _serverIP = value; }
        }

        private string _userName;
        /// <summary>
        /// 用户名
        /// </summary>
        public string UserName
        {
            get { return _userName; }
            set { _userName = value; }
        }

        private string _password;
        /// <summary>
        /// 密码
        /// </summary>
        public string Password
        {
            get { return _password; }
            set { _password = value; }
        }

        private string _subDirectory;
        /// <summary>
        /// 目录
        /// </summary>
        public string SubDirectory
        {
            get { return _subDirectory; }
            set { _subDirectory = value; }
        }
    }

}
